package com.wasp.collector.service.akka;

import com.wasp.collector.data.TimeInterval;

import akka.actor.AbstractActor;
import akka.actor.ActorRef;
import akka.actor.Props;
import akka.japi.pf.ReceiveBuilder;
import akka.routing.ConsistentHashingRouter.ConsistentHashableEnvelope;
import akka.routing.FromConfig;
import scala.PartialFunction;
import scala.runtime.BoxedUnit;

public class StatsService extends AbstractActor {
	private ActorRef computerRouter = computerRouter = getContext().actorOf(FromConfig.getInstance().props(Props.create(ComputerWorker.class)),
			"computerRouter");
	ActorRef collectWorkerRouter = getContext().actorOf(
		      FromConfig.getInstance().props(Props.create(CollectWorker.class)),
		      "collectWorkerRouter");
	private ActorRef defaultRouter = getContext().actorOf(FromConfig.getInstance().props(Props.create(DefaultWorker.class)),
			"defaultRouter");
	
	public StatsService(){
		
	}

	@Override
	public PartialFunction<Object, BoxedUnit> receive() {
		return ReceiveBuilder.match(HttpClientActorContext.class, job -> {
			collect(job);
		})
		.match(TimeInterval.class, interval->{
			collect(interval);
		}).build();
	}
	
	private void collect(HttpClientActorContext context){
		collectWorkerRouter.tell(new ConsistentHashableEnvelope(context,context), sender());
	}
	private void collect(TimeInterval interval){
		getContext().actorSelection("/user/statsService/collectWorkerRouter/*").tell(interval, sender());;
	}
}
